/*******************************
 * utf-8
 *
 * sphinx http api
 *
 * zhaigy@ucweb.com 2011-04
 *
 * *****************************/
//TODO 搜索功能补全
//TODO 错误异常返回值检查
//TODO 配置文件，记录日志
//TODO 统计、调试功能    
//TODO 连接池 
//TODO 增加信息、统计、调试页面 
//TODO 调试
#include <cctype>
#include <cstring>
#include <iostream>
#include <fstream>
#include <string>
#include <list>
#include <sstream>
#include <algorithm>
#include <functional>

#include <event2/http.h>
#include <event2/http_struct.h>
#include <event2/thread.h>
#include <event2/event.h>
#include <event2/event_struct.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>

#include "../libsphinxclient/sphinxclient.h"
#include "../libthreadpool/CThread.h"
#include "../json/json/json.h"
#include "../log/log.h"

#include "utils.h"
#include "Config.h"
#include "SphinxQueryData.h"
#include "SearchResult.h"

using namespace std;

/**
 * 记录用户的输入参数
 */
struct GlobalParameter
{
	GlobalParameter():  is_daemon(false),
                        host("0.0.0.0"),
	                    port(2046), 
                        proxyed_host("localhost"), 
                        proxyed_port(9312),
                        pid_path_name("./logs/proxy.pid"),
                        log_path_name("./logs/proxy.log"),
                        thread_pool_min(3),
                        thread_pool_max(100),
                        log_level("INFO")
    {
	} 
    bool is_daemon;         //是否是后台执行
	string host;		    //ip
	int port;		        //端口
	string proxyed_host;    //代理的sphinx的ip
	int proxyed_port;	    //dialing的sphinx的端口
    string log_path_name;   //日志文件
    string pid_path_name;   //PID文件
    int thread_pool_min;
    int thread_pool_max;
    string log_level;       //日志等级
};

/**
 * 记录程序运行统计信息
 */
struct Statistic
{
    Statistic():query_count(0),
                query_error_count(0),
                start_time(time(NULL))
    {
    }
	volatile long long query_count;          //查询总次数
	volatile long long query_error_count;	//查询出错次数
	time_t start_time;	        //启动时间
};


struct http_connection
{
	int fd;
	struct evbuffer *inbuf;
	struct evbuffer *outbuf;
	
	LIST_ENTRY( http_connection ) next;
};
/**
 * 自定义结构保存client信息
 */
struct sockclient 
{
    char ip[50]; 
    struct http_connection conn;
    struct event * p_read_ev;
    struct event * p_write_ev;
    struct event_base * p_base;
};

/********************全局变量**********************/
//注意：以下两个都是线程级别的，并非真正的全局变量，这里是暂时的
static struct event_base * gp_event_base                   = NULL;
static Statistic * gp_statistic                            = NULL;
static GlobalParameter * gp_parameter                      = NULL;
static CThreadPool * gp_thread_pool                        = NULL;
static map<pthread_t,sphinx_client *> g_sphinx_client_map;
//sphinx client池的互斥锁
static pthread_mutex_t g_sphinx_client_mutex               = PTHREAD_MUTEX_INITIALIZER;
/**************************************************/
CLog * gp_log=NULL;
#define ERROR_LOG(fmt...) \
    do {if(gp_log) gp_log->error(fmt);}while(0)
#define WARN_LOG(fmt...)  \
    do {if(gp_log) gp_log->warn(fmt);}while(0)
#define INFO_LOG(fmt...)  \
    do {if(gp_log) gp_log->info(fmt);}while(0)
//#define INFO_LOG(fmt...) 
#define DEBUG_LOG(fmt...) \
    do {if(gp_log) gp_log->debug(fmt);}while(0)
//#define DEBUG_LOG(fmt...) 
/**************************************************/

/**
 * 实现从输入到查询数据的初步转换
 */
int parserQuery(evkeyvalq & http_query,QueryData & qd)
{
	//已经处理全部预定义Key
	for (evkeyval * evkv = http_query.tqh_first, *evkvEnd = *http_query.tqh_last; evkv != 0; evkv = evkv->next.tqe_next) 
    {
		DEBUG_LOG("evkv->key:%s, evkv->value:%s.", evkv->key, evkv->value);
		if (strcmp("index", evkv->key) == 0) 
        {
			qd.index = evkv->value;
		} 
        else if (string("q") == evkv->key) 
        {
			qd.q = evkv->value;
		} 
        else if (string("filter") == evkv->key) 
        {
			qd.filters.push_back(evkv->value);
		} 
        else if (string("idrange") == evkv->key) 
        {
			qd.idrange = evkv->value;
		} 
        else if (string("start") == evkv->key) 
        {
			qd.start = atoi(evkv->value);
		} 
        else if (string("rows") == evkv->key) 
        {
			qd.rows = atoi(evkv->value);
		} 
        else if (string("fieldweights") == evkv->key) 
        {
			qd.fieldweights = evkv->value;
		} 
        else if (string("retries") == evkv->key) 
        {
			qd.retries = atoi(evkv->value);
		} 
        else if (string("maxquerytime") == evkv->key) 
        {
			qd.maxquerytime = atoi(evkv->value);
		} 
        else if (string("connecttimeout") == evkv->key) 
        {
			qd.connecttimeout = atoi(evkv->value);
		} 
        else if (string("groupby") == evkv->key) 
        {
			qd.groupby = evkv->value;
		} 
        else if (string("groupdistinct") == evkv->key) 
        {
			qd.groupdistinct = evkv->value;
		} 
        else if (string("matchmode") == evkv->key) 
        {
			qd.matchmode = evkv->value;
		} 
        else if (string("rankingmode") == evkv->key) 
        {
			qd.rankingmode = evkv->value;
		} 
        else if (string("sortmode") == evkv->key) 
        {
			qd.sortmode = evkv->value;
		} 
        else if (string("select") == evkv->key) 
        {
			qd.select = evkv->value;
		} 
        else 
        {
			WARN_LOG("不支持的输入值,key:%s, value:%s", evkv->key, evkv->value);
		}

		if (evkv == evkvEnd) 
        {
			break;
		}
	}
	return 0;
}

void disconnect_conn (struct http_connection *conn ){
	shutdown( conn->fd, SHUT_RDWR );
	close(conn->fd);
	if(conn->inbuf) evbuffer_free( conn->inbuf );
	if(conn->outbuf) evbuffer_free( conn->outbuf );
}

/**
 * 创建Sphinx Client,并打开长连接.
 */
static sphinx_client * create_sphinx_client()
{   
    DEBUG_LOG("%s",__FUNCTION__);
    sphinx_client * client = sphinx_create(SPH_TRUE);
    if(client != NULL)
    {
        if( sphinx_set_server(client, gp_parameter->proxyed_host.c_str(),gp_parameter->proxyed_port)
            &&
            sphinx_open(client) )
        {
            return client;
        }
        //默认1秒钟超时
        //sphinx_set_connect_timeout(client, 5);
        WARN_LOG("sphinx open fail:%s",sphinx_error(client));
        sphinx_destroy(client);
    }
    return NULL;
}

/**
 * 加锁，获取有效的Sphinx Client。
 */
sphinx_client * get_sphinx_client(pthread_t tid)
{
    DEBUG_LOG("%s,tid=%u",__FUNCTION__,tid);
    pthread_mutex_lock(&g_sphinx_client_mutex);
    map<pthread_t,sphinx_client *>::iterator it = g_sphinx_client_map.find(tid);
    sphinx_client * client = NULL;
    if(it != g_sphinx_client_map.end() )
    {
        client = (*it).second;
    }
    else 
    {
        client = create_sphinx_client();
        if( client != NULL ) 
        {
            g_sphinx_client_map[tid]=client;
            DEBUG_LOG("create sphinx client:tid=%u,client pool num=%d,client=%u",tid,g_sphinx_client_map.size(),client);
        }
    }
    pthread_mutex_unlock(&g_sphinx_client_mutex);
    return client;
}

sphinx_client * get_thread_sphinx_client()
{
    pthread_t tid = pthread_self();
    sphinx_client * client = get_sphinx_client(tid);
    return client;
}

int thread_pool_before_run_task_cb(pthread_t tid)
{
    sphinx_client * client = get_sphinx_client(tid);
    return client != NULL ? 0:1;
}

/**
 * 释放并关闭一个sphinx client
 */
int release_sphinx_client(pthread_t tid)
{
    pthread_mutex_lock(&g_sphinx_client_mutex);
    map<pthread_t,sphinx_client *>::iterator it = g_sphinx_client_map.find(tid);
    sphinx_client * client = NULL;
    if(it != g_sphinx_client_map.end())
    {
        client = it->second;
        g_sphinx_client_map.erase(it);
    }
    pthread_mutex_unlock(&g_sphinx_client_mutex);
    if(client)
    {
        sphinx_destroy(client);
        INFO_LOG("release_sphinx_client,tid=%u,client=%u",tid,client);
    }
    else
    {
        WARN_LOG("[%u] why client is null ?",tid);
    }
    return 0;
}

void release_thread_sphinx_client()
{
    pthread_t tid = pthread_self();
    release_sphinx_client(tid);
} 

/**
 * 调用客户端接口查询
 * TODO 重点！！！
 */
void search(const SphinxQueryData & sqd, SearchResult * sr)
{
    sphinx_client * const client = get_thread_sphinx_client();
    if( client == NULL )
    {
        throw "sphinx client is NULL";
    }
	sphinx_reset_filters(client);
	sphinx_reset_groupby(client);

	sr->set_client(client);

	//filter - 属性值，属性范围，属性范围float,id范围（仅可以设置一个值）
    bool set_ok=false;
    do 
    {
        if (!sqd.filters.empty()) 
        {
            for (list < SphinxQueryData::Filter >::const_iterator it = sqd.filters.begin(); it != sqd.filters.end(); ++it) 
            {
                const SphinxQueryData::Filter & filter = *it;
                if (filter.type == SphinxQueryData::Filter::kValue) 
                {
                    if (sphinx_add_filter (client, filter.attr_name.c_str(),
                                filter.values_num, filter.values, filter.exclude) != SPH_TRUE) 
                    {
                        break;
                    }
                } 
                else if (filter.type == SphinxQueryData::Filter::kIntRange) 
                {
                    if (sphinx_add_filter_range (client, filter.attr_name.c_str(), 
                                filter.int_min, filter.int_max, filter.exclude) != SPH_TRUE) 
                    {
                        break;
                    }
                } 
                else if (filter.type == SphinxQueryData::Filter::kFloatRange) 
                {
                    if (sphinx_add_filter_float_range (client, 
                                filter.attr_name.c_str(), filter.float_min, filter.float_max, filter.exclude) != SPH_TRUE) 
                    {
                        break;
                    }
                } 
                else 
                {
                    throw "Filter 未知的设置类型";
                }
            }
        }
        if (sqd.idrange_min > 0L) 
        {	//id=0 这个在sphinx中是不合法的
            if (sphinx_set_id_range (client, sqd.idrange_min, sqd.idrange_max) != SPH_TRUE) 
            {
                break;
            }
        }
        //第4个参数是否可以设置为rows*10000?
        if (sphinx_set_limits (client, sqd.start, sqd.rows, sqd.start + sqd.rows, 0) != SPH_TRUE) 
        {
            break;
        }
        //fieldweight - 字段数，字段名数组，权重数组（整数）
        if (sqd.fieldweights_num > 0) 
        {
            if (sphinx_set_field_weights(client, sqd.fieldweights_num, 
                        sqd.fieldweights_fields, sqd.fieldweights_weights) != SPH_TRUE) 
            {
                break;
            }
        }

        if (sqd.retries > 0) 
        {
            //间隔毫秒
            if (sphinx_set_retries(client, sqd.retries, 100) != SPH_TRUE) 
            {
                break;
            }
        }

        if (sqd.maxquerytime > 0) 
        {
            //毫秒
            if (sphinx_set_max_query_time(client, sqd.maxquerytime) != SPH_TRUE) 
            {
                break;
            }
        }
        if (sqd.connecttimeout > 0) 
        {
            //转换为秒
            if (sphinx_set_connect_timeout (client, float (sqd.connecttimeout) / 1000) != SPH_TRUE) 
            {
                break;
            }
        }
        //groupby - groupfun , sort字符串
        if (!sqd.groupby_attr.empty()) 
        {
            if (sphinx_set_groupby(client, sqd.groupby_attr.c_str(), sqd.groupby_func, sqd.groupby_sort.c_str()) != SPH_TRUE) 
            {
                break;
            }
            if (!sqd.groupdistinct.empty()) 
            {
                if (sphinx_set_groupby_distinct (client, sqd.groupdistinct.c_str()) != SPH_TRUE) 
                {
                    break;
                }
            }
        }

        if (sphinx_set_match_mode(client, sqd.matchmode) != SPH_TRUE
                || sphinx_set_ranking_mode(client, sqd.rankingmode) != SPH_TRUE
                || sphinx_set_sort_mode(client, sqd.sortmode,
                    sqd.sortclause.c_str()) != SPH_TRUE) 
        {
            break;
        }

        if (!sqd.select.empty()) 
        {
            if (sphinx_set_select(client, sqd.select.c_str()) != SPH_TRUE) 
            {
                break;
            }
        }
       set_ok=true; 
    } while(0);
    if(!set_ok)
    {
        throw sphinx_error(client);
    }
	sphinx_result *res = sphinx_query(client, sqd.q.c_str(),sqd.index.c_str(), NULL);
    if(res == NULL)
    {
        WARN_LOG("query error:%s",sphinx_error(client));
        release_thread_sphinx_client();
        //可能存在错误的连接
        throw sphinx_error(client);
    }
	sr->set_result(res);
}

/**使用jsoncpp实现的转换*/
string convert_to_json_string(const SearchResult & sr)
{
	Json::Value root;
	//拼输出
    root["status_info"]="0=SEARCHD_OK, 1=SEARCHD_ERROR, 2=SEARCHD_RETRY, 3=SEARCHD_WARNING";
    root["status"] = sr.getStatus();
    if(sr.getStatus() != SEARCHD_OK)
    {
        WARN_LOG("search result status:%d",sr.getStatus());
    }
    if (sr.getStatus() == SEARCHD_ERROR) 
    {
        root["error"] = sr.getError();
        WARN_LOG("error=[%s]",sr.getError());
    } 
    else if (sr.getStatus() == SEARCHD_RETRY) 
    {
        //无任何说明
    } 
    else //ok warning
    {
        if (sr.getStatus() == SEARCHD_WARNING) 
        {
            root["warning"] = sr.getWarning();
        }
        root["total"]     = sr.getTotal();
        root["time"]      = sr.getTime();
        int matchNumb     = sr.getMatchNumb();
        root["matchNumb"] = matchNumb;
        //匹配集合 root["matches"];
        for (int i = 0; i < matchNumb; ++i) 
        {
            Json::Value match;
            match["id"] = (long long)sr.getId(i);
            int attrNumb = sr.getAttrNumb();
            for (int j = 0; j < attrNumb; ++j) 
            {
                if (sr.getAttrType(j) == (SPH_ATTR_MULTI | SPH_ATTR_INTEGER)) 
                {
                    unsigned int *mva = sr.getAttrMultiIntValue(i, j);
                    Json::Value array_temp;
                    for (unsigned int k = 0; k < mva[0]; ++k) 
                    {
                        array_temp.append(mva[1 + k]);
                    }
                    match[sr.getAttrName(j)] = array_temp;
                } 
                else if (sr.getAttrType(j) == SPH_ATTR_FLOAT) 
                {
                    match[sr.getAttrName(j)] = sr.getAttrFloatValue(i, j);
                } 
                else //int 
                {	
                    match[sr.getAttrName(j)] = sr.getAttrIntValue(i, j);
                }
            }
            //这里添加进来
            root["matches"].append(match);
        }

        //for words
        int wordNumb = sr.getWordNumb();
        root["wordNumb"] = wordNumb;
        //words
        for (int i = 0; i < wordNumb; ++i) 
        {
            Json::Value word;
            word["word"] = sr.getWord(i);
            word["hits"] = sr.getHits(i);
            word["docs"] = sr.getDocs(i);
            root["words"].append(word);
        }
    }
    Json::StyledWriter writer;
	return writer.write(root);
}

/**
 * 构造一个简单的错误信息的Json字符串
 */
string make_inner_error_json_string(string e)
{
	Json::Value root;
	root["status_info"] = "0=SEARCHD_OK, 1=SEARCHD_ERROR; 2=SEARCHD_RETRY; 3=SEARCHD_WARNING";
	root["status"]      = SEARCHD_ERROR;
	root["error"]       = e;
    Json::StyledWriter json_writer;
	return json_writer.write(root);
}

/**
 * 返回结果必须是json格式，不能抛出异常
 */
bool deal_search_handler(struct evkeyvalq & http_query,string & json)
{
	try 
    {
        QueryData query_data;
		if(parserQuery(http_query,query_data) != 0 || query_data.q.empty())
        {
			ERROR_LOG("查询语句是空的");
            json = make_inner_error_json_string("查询语句是空的");
            return false;
		}
        //转换为sphinx可用的查询数据格式
		SphinxQueryData sqd(query_data);
		SearchResult searchResult;
        search(sqd, &searchResult);
        json = convert_to_json_string(searchResult);
		return true;
	}
	catch(const char *e) 
    {
		ERROR_LOG("异常:%s", e);
		json = make_inner_error_json_string(string("异常：") + e);
	}
	catch(const string & e) 
    {
		ERROR_LOG("异常:%s", e.c_str());
		json = make_inner_error_json_string("异常：" + e);
	}
	catch(...) 
    {
		ERROR_LOG("未知的异常");
		json = make_inner_error_json_string("未知的异常");
	}
    return false;
}

/**
 * 处理状态请求,不允许抛出异常
 */
bool deal_status_handler(string &json)
{
    DEBUG_LOG("%s",__FUNCTION__);
	Json::Value root;
	root["status_info"] = "0=SEARCHD_OK, 1=SEARCHD_ERROR; 2=SEARCHD_RETRY; 3=SEARCHD_WARNING";
 
    sphinx_client * const client = get_thread_sphinx_client();
    if( client == NULL )
    {
        const char * err = "sphinx client 异常";
        json = make_inner_error_json_string(err);
        return false;
    }

	int row_num=0;
	int col_num=0;
	char **status = sphinx_status(client, &row_num, &col_num);
	if (!status) 
    {
		root["status"] = SEARCHD_ERROR;
		root["error"]  = sphinx_error(client);
	} 
    else 
    {
		root["status"] = SEARCHD_OK;
		for (int i = row_num * col_num - 1; i >= 0; i -= 2) 
        {
			root[status[i - 1]] = status[i];
		}
	}
	sphinx_status_destroy(status, row_num, col_num);

    Json::StyledWriter json_writer;
    json = json_writer.write(root);
	return true;
}

/**
 * 处理统计请求
 */
string deal_statistic_handler(void)
{
	Json::Value root;
    root["query_count"]       = gp_statistic->query_count;
    root["query_error_count"] = gp_statistic->query_error_count;
    root["start_time"]        = ctime( &gp_statistic->start_time );
    Json::StyledWriter json_writer;
    string json = json_writer.write(root);
    return json;
}

static long get_file_size( FILE *fp )
{
    long file_size ;
    fseek( fp, 0, SEEK_END );
    file_size = ftell( fp );
    fseek( fp, 0, SEEK_SET );
    return file_size;
}

/**
 * 工具方法：读取文本文件内容到内存,文本文件要求是小的，最大不能操作1M，超过仅返回1M内容。
 */
int read_small_text_file( const char * file_name, struct evbuffer * buf) 
{
    string file("./web/");
    file += file_name;
    FILE * fp=fopen(file.c_str(),"rb");
    if(fp==NULL)
    {
        return -1;
    }
    long file_size=get_file_size(fp);
    DEBUG_LOG("file:%s, size:%ld",file.c_str(),file_size);
    int size=-1;
    if(file_size < 1024L*1024)
    {
        char * data = (char*) malloc( (int)file_size );
        if(data != NULL)
        {
            size=fread( data, 1,file_size, fp );
            if(size>0)
            {
                evbuffer_add( buf, data, size );
            }
            free( data );
        }
    }
    fclose(fp);
    DEBUG_LOG("%s:%d",__FUNCTION__,size);
    return size;
}

//void dumb (struct evhttp_request *req, void *arg) 
//{
//    struct evbuffer* buf = evbuffer_new();
//    evhttp_send_reply(req, HTTP_OK, "OK", buf);
//    return;
//}

#define RESP_HEAD(buf,status,status_str,keep_alive,data_length) do {\
    evbuffer_add_printf( buf, "HTTP/1.1 %d %s\r\n", status, status_str ); \
    evbuffer_add_printf( buf, "Server: sphinx_http_proxy\r\n" ); \
    add_time_header( buf ); \
    evbuffer_add_printf( buf, "Content-Type: text/plain; charset=UTF-8\r\n" ); \
    evbuffer_add_printf( buf, "Content-Length: %d\r\n", data_length ); \
    if( keep_alive) { \
        evbuffer_add_printf( buf, "Connection: keep-alive\r\n" ); \
    } else { \
        evbuffer_add_printf( buf, "Connection: close\r\n" ); \
    } \
    evbuffer_add( buf, "\r\n", 2 ); \
}while(0)

void set_http_response( struct evbuffer *buf, int status, const char *status_str,
        bool keep_alive, const char *more_info,int data_length )
{
	RESP_HEAD(buf,status,status_str,keep_alive,data_length);
	evbuffer_add( buf, more_info, data_length );
}

void set_http_response_buf( struct evbuffer *buf, int status, const char *status_str,
        bool keep_alive, struct evbuffer * body_data)
{
    int data_length = evbuffer_get_length(body_data);
	RESP_HEAD(buf,status,status_str,keep_alive,data_length);
	evbuffer_add_buffer( buf, body_data );
}
#undef RESP_HEAD

/**
 * 检查请求是否有效
  @return 有效返回0.
*/
int check_request_valid( struct evbuffer *buf, const struct http_request *request )
{
    if( request->type == HTTP_UNKNOWN )
    {
        return -1;
    }

    /* HTTP 1.1 needs Host header */
    if( request->ver.major == 1 && request->ver.minor == 1 )
    {
        const char *host = http_get_header_value( request->headers, "Host" );
        if( host == 0 )
        {
            return -2;
        }
    }
    return 0;
}

/**
 * 处理器,注意,此函数内部如无系统崩溃，不会报错，所有异常应控制在本函数内部
 */
void http_handler(struct sockclient * socket_client, bool * keep_alive)
{  
    struct http_request *request = NULL;
    try 
    {
        struct http_connection *conn = &socket_client->conn;
        struct evbuffer *buf = conn->outbuf;
        request = http_request_parse( conn->inbuf );
        if(request == NULL)
        {
            const char * info = "ERROR: malloc fail";
            WARN_LOG("%s",info);
            set_http_response ( buf, HTTP_BADREQUEST, CODE_STR( HTTP_BADREQUEST ), false, info, strlen(info) );
            goto BUF_OUT;
        }

        //for goto cross variant
        {
            //对是否保持连接进行处理
            *keep_alive = false;
            http_header * tmp=NULL;
            TAILQ_FOREACH( tmp, request->headers, next)
            {
                if( strcmp(tmp->name,"Connection")==0)
                {
                    if(strcasecmp(tmp->value,"keep-alive")==0)//Keep-Alive
                    {
                        *keep_alive = true;
                    }
                    break;
                }
            }
            
            {
                int ret = check_request_valid( buf, request );
                if(  ret == -1 )
                {
                    const char * info = "ERROR: Not Implemented Method";
                    WARN_LOG("%s",info);
                    set_http_response( buf, HTTP_NOTIMPLEMENT, CODE_STR( HTTP_NOTIMPLEMENT ),*keep_alive,info,strlen(info) );
                    goto BUF_OUT;
                } 
                else if(ret == -2)
                {
                    const char * info = "ERROR: Bad Request: needs Host header";
                    WARN_LOG("%s",info);
                    set_http_response ( buf, HTTP_BADREQUEST, CODE_STR( HTTP_BADREQUEST ),*keep_alive, info, strlen(info) );
                    goto BUF_OUT;
                }
            }

            struct evhttp_uri * uri= evhttp_uri_parse(request->uri);
            if(uri == NULL )
            {
                const char * info = "ERROR: uri parse fail";
                WARN_LOG("%s",info);
                set_http_response ( buf, HTTP_BADREQUEST, CODE_STR( HTTP_BADREQUEST ),*keep_alive, info, strlen(info) );
                goto BUF_OUT;
            }

            string path = evhttp_uri_get_path(uri);
            DEBUG_LOG("path:%s", path.c_str());
            evhttp_uri_free(uri);

            evbuffer * body_data = evbuffer_new();
            if( body_data == NULL )
            {
                const char * info = "ERROR: malloc fail";
                set_http_response( buf, HTTP_BADREQUEST, CODE_STR( HTTP_BADREQUEST ),*keep_alive, info, strlen(info) );
                goto BUF_OUT;
            }
            
            bool is_ok = false;
            path="/status";
            if (path == "/search") 
            {	
                //处理检索
                gp_statistic->query_count++;
                struct evkeyvalq http_query;
                
                if(evhttp_parse_query(request->uri,&http_query) != 0)
                {
                    const char * info = "ERROR: parse uri fail";
                    WARN_LOG("%s",info);
                    set_http_response ( buf, HTTP_BADREQUEST, CODE_STR( HTTP_BADREQUEST ),*keep_alive, info, strlen(info) );
                    goto BUF_OUT; 
                }

                string json;
                is_ok = deal_search_handler(http_query,json);
                evhttp_clear_headers(&http_query);
                const char * data = json.c_str();
                evbuffer_add(body_data,data,strlen(data));
            } 
            else 
            {
                if (path == "/status") 
                {
                    //代理的统计信息
                    string json;
                    is_ok = deal_status_handler(json);
                    const char * data = json.c_str();
                    evbuffer_add(body_data,data,strlen(data));
                } 
                else if (path == "/statistic") 
                {
                    string json = deal_statistic_handler();
                    const char * data = json.c_str();
                    evbuffer_add(body_data,data,strlen(data));
                    is_ok=true;
                } 
                else if (path == "/" || path == "") 
                {
                    int res = read_small_text_file("info.html",body_data);
                    if(res > 0)
                    {
                        is_ok=true;
                    }
                } 
                else if (path == "/debug") 
                {
                    int res = read_small_text_file("debug.html",body_data);
                    if( res > 0)
                    {
                        is_ok=true;
                    }
                } 
                else 
                {
                    WARN_LOG("未知的路径");
                    //不支持的请求
                    const char * info = "ERROR: HTTP_NOTIMPLEMENT";
                    set_http_response ( buf, HTTP_NOTIMPLEMENT, CODE_STR( HTTP_NOTIMPLEMENT ),
                            *keep_alive, info, strlen(info) );
                    goto BUF_OUT;
                }
            }
            if(is_ok)
            {
                INFO_LOG("[%s] OK", path.c_str());
                set_http_response_buf(buf,HTTP_OK,CODE_STR( HTTP_OK ),*keep_alive,body_data);
            }
            else
            {
                WARN_LOG("[%s] Fail",path.c_str());
                *keep_alive = false;
                set_http_response_buf(buf,HTTP_BADREQUEST,CODE_STR( HTTP_BADREQUEST ),*keep_alive,body_data);
            }
            evbuffer_free(body_data);
        }
    BUF_OUT:
        //输出数据
        if( evbuffer_get_length( buf ) > 0 ) 
        {
            DEBUG_LOG("out put response ..."); 
            evbuffer_write( buf, conn->fd ); 
        }
        if( request )  http_request_free( request );
        return;
    }
    //FIXME:需要更稳妥的处理异常
    catch(string err) 
    {
        ERROR_LOG(err.c_str());
	}
	catch(...) 
    {
		ERROR_LOG("未知错误");
	}
	gp_statistic->query_error_count++;
    if( request ) http_request_free(request);
}

/**
 * 加载配置文件
 */
bool load_config_file(){
	try 
    {
		const char * config_file = "./etc/proxy.ini";
		Config configSettings(config_file);

		gp_parameter->is_daemon       = configSettings.Read("is_daemon", gp_parameter->is_daemon);
		gp_parameter->host            = configSettings.Read("host", gp_parameter->host);
		gp_parameter->port            = configSettings.Read("port", gp_parameter->port);
		gp_parameter->proxyed_host    = configSettings.Read("proxyed_host", gp_parameter->proxyed_host);
		gp_parameter->proxyed_port    = configSettings.Read("proxyed_port", gp_parameter->proxyed_port);
		gp_parameter->pid_path_name   = configSettings.Read("pid_path_name", gp_parameter->pid_path_name);
		gp_parameter->log_path_name   = configSettings.Read("log_path_name", gp_parameter->log_path_name);
		gp_parameter->thread_pool_min = configSettings.Read("thread_pool_min", gp_parameter->thread_pool_min);
		gp_parameter->thread_pool_max = configSettings.Read("thread_pool_max", gp_parameter->thread_pool_max);
		gp_parameter->log_level = configSettings.Read("log_level", gp_parameter->log_level);
		return true;
	}
	catch(Config::File_not_found fnf) 
    {
		fprintf(stderr,"配置文件没有找到\n");
	}
	catch(...) 
    {
		fprintf(stderr,"未知错误\n");
	}
	return false;
}

/**
 * 检查参数配置，失败就退出程序
 */
void check_config_parameter()
{
	cout << "is_daemon: " << gp_parameter->is_daemon << endl;
    cout << "host: " << gp_parameter->host << ":" << gp_parameter-> port << endl;
	cout << "proxyed_host: " << gp_parameter-> proxyed_host << ":" << gp_parameter->proxyed_port << endl;

	if (gp_parameter->host.empty() || gp_parameter->proxyed_host.empty()) 
    {
		cerr << "配置不正确" << endl;
		exit(1);
	}
}

void ExecSafeExit(int aiSignal)
{
	struct timeval tv;
	tv.tv_sec = 1;
	tv.tv_usec = 0;
    event_base_loopexit(gp_event_base,&tv);
}

/**
 * 处理终端消息通知
 */
void SetSignalHandle(void)
{
	signal(SIGALRM, SIG_IGN);
	signal(SIGPIPE, SIG_IGN);
	signal(SIGUSR2, SIG_IGN);
	signal(SIGHUP, SIG_IGN);
	signal(SIGQUIT, SIG_IGN);
	signal(SIGUSR1, SIG_IGN);
	signal(SIGINT, SIG_IGN);
	signal(SIGTERM, SIG_IGN);

	signal(SIGINT, ExecSafeExit);
	signal(SIGTERM, ExecSafeExit);
	signal(SIGQUIT, ExecSafeExit);
}

void check_pid_file()
{
    if(access(gp_parameter->pid_path_name.c_str(),F_OK)==0)
    {
        ERROR_LOG("PID[%s]文件已经存在",gp_parameter->pid_path_name.c_str());
        exit(1);
    }
    pid_t pid=getpid();
    FILE * pidFile=fopen(gp_parameter->pid_path_name.c_str(),"w");
    if(pidFile==NULL)
    {
        ERROR_LOG("PID文件创建失败:%s",strerror(errno));
        exit(1);
    }
    fprintf(pidFile,"%d",pid);
    fclose(pidFile);
}

void delet_pid_file()
{
    if(remove(gp_parameter->pid_path_name.c_str())<0)
    {
        WARN_LOG("PID[%s]文件删除失败",gp_parameter->pid_path_name.c_str());
    }
}

//函数：设置sock为non-blocking mode
int setSockNonBlock(int sock) 
{
    int flags = fcntl(sock, F_GETFL, 0);
    if (flags < 0) 
    {
        return -1;
    }
    if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0) 
    {
        return -1;
    }
}

void send_bad_req(int fd)
{
    const char * bad_msg="error query";
    send(fd,bad_msg,strlen(bad_msg),0);
}

/**
 * 客户端的sock是可以重用的
 * 回调函数：从client端读
 */
void read_callback(int fd, short ev, void *arg) {
    DEBUG_LOG("%s:%s:fd=%d,ev=%d",__FILE__,__FUNCTION__,fd,ev);
    struct sockclient *socket_client = (struct sockclient *)arg;
    struct http_connection *conn = &socket_client->conn;

    conn->inbuf = evbuffer_new();
    conn->outbuf = evbuffer_new();
     
    int ret = evbuffer_read( conn->inbuf, fd, 4096 );
    DEBUG_LOG("buffer read ret=%d",ret);
    if( ret <= 0 )
    {
#define CLOSE_CONN do { \
    event_del(socket_client->p_read_ev);\
    DEBUG_LOG("event_base_loopexit");\
    evbuffer_free(conn->inbuf);\
    evbuffer_free(conn->outbuf);\
    conn->inbuf=NULL;\
    conn->outbuf=NULL;\
    event_base_loopexit(socket_client->p_base,NULL);\
}while(0)
        CLOSE_CONN;
        return;
    }

    bool keep_alive = false;
    http_handler(socket_client,&keep_alive);
    if(!keep_alive)
    {
        CLOSE_CONN;
#undef CLOSE_CONN
    }
    else
    {
        evbuffer_free(conn->inbuf);
        evbuffer_free(conn->outbuf);
        conn->inbuf=NULL;
        conn->outbuf=NULL;
        DEBUG_LOG("end read_callback");
    }
}

class SimpleWorkTask: public CTask 
{
    public:
        SimpleWorkTask(struct sockclient * socket_client)
        {
            this->socket_client = socket_client;
        }

        int Run() 
        {
            DEBUG_LOG("%s","SimpleWorkTask Run: begin...");
            struct http_connection *conn = &socket_client->conn;
            DEBUG_LOG("sock fd=%d",conn->fd);
            //还不能处理成非阻塞的，掌握的知识还不够
            //setSockNonBlock(conn->fd);
            //必须显式设置为阻塞
            fcntl ( conn->fd, F_SETFL, 0 );

            for(;1;)
            {
                conn->inbuf = evbuffer_new();
                conn->outbuf = evbuffer_new();
                
                bool keep_alive = false;
                int ret = evbuffer_read( conn->inbuf, conn->fd, 4096 );
                DEBUG_LOG("evbuffer_read:%d",ret);
                if(ret > 0){
                    http_handler(this->socket_client, &keep_alive);
                }
               
                evbuffer_free(conn->inbuf);conn->inbuf=NULL;
                evbuffer_free(conn->outbuf);conn->outbuf=NULL;
               
                if( ret <= 0 ){
                    break;
                } 
                
                if(!keep_alive)
                {
                    break;
                }
            }
            disconnect_conn(conn);
            free(socket_client); 
            DEBUG_LOG("%s","SimpleWorkTask Run: end");
            return 0;
        }
    private:
        struct sockclient * socket_client;
};

/**
 * 旧的实现方式，留待备用，需要对libevent更好的掌握
 */
class CWorkTask2: public CTask 
{
    public:
        CWorkTask2(struct sockclient * socket_client)
        {
            this->socket_client = socket_client;
        }

        int Run() 
        {
            DEBUG_LOG("%s","CWorkTask2 Run: begin...");
            struct http_connection *conn = &socket_client->conn;
            DEBUG_LOG("sock fd=%d",conn->fd);

            //setSockNonBlock(conn->fd);
            //不知道为什么，如下的代码是不行的
            socket_client->p_base = event_base_new();
            if(socket_client->p_base == NULL)
            {
                WARN_LOG("base is null");
                return -1;
            }
            DEBUG_LOG("event_new : fd=%d",conn->fd);
            socket_client->p_read_ev = event_new(socket_client->p_base,conn->fd,
                    EV_READ|EV_PERSIST,read_callback,socket_client);
            event_add(socket_client->p_read_ev,NULL);
            DEBUG_LOG("%s","CWorkTask2 Run : event_base_dispatch");
            event_base_dispatch(socket_client->p_base);
            DEBUG_LOG("%s","CWorkTask2 Run : dispatch end");


            //free
            event_free(socket_client->p_read_ev);
            socket_client->p_read_ev = NULL;
            event_base_free(socket_client->p_base);
            socket_client->p_base = NULL;
            disconnect_conn(&socket_client->conn);
            free(socket_client); 
            DEBUG_LOG("%s","CWorkTask2 Run: end");
            return 0;
        }
    private:
        struct sockclient * socket_client;
};


/**
 * 旧的实现方式，留待备用
 */
void accept_callback(int fd, short ev, void *arg) 
{
    DEBUG_LOG("%s:%s",__FILE__,__FUNCTION__);
    //接受client连接请求
    struct sockaddr_in client_addr;
    socklen_t client_addr_len = sizeof(client_addr);

    int conn_sock= accept(fd, (struct sockaddr *)&client_addr, &client_addr_len);
    if (conn_sock == -1) 
    {
        WARN_LOG("accept failed:%s",strerror(errno));
        return;
    }

    //保存client端信息到自定义结构体中
    struct sockclient * socket_client = (struct sockclient*)malloc(sizeof(struct sockclient));
    if (socket_client == NULL) 
    {
        WARN_LOG("malloc failed");
        shutdown(conn_sock,SHUT_RDWR);
        return;
    }

    if (!inet_ntop(AF_INET, &(client_addr.sin_addr), socket_client->ip, sizeof(socket_client->ip))) {
        WARN_LOG("inet_ntop failed");
        shutdown(conn_sock,SHUT_RDWR);
        return;
    }

    DEBUG_LOG("sock: %d, from: %s", conn_sock, socket_client->ip);
    struct http_connection *conn = &socket_client->conn;
    conn->fd = conn_sock;
    conn->inbuf = NULL;
    conn->outbuf = NULL;
    
    //CWorkTask2 * task = new CWorkTask2(socket_client);
    SimpleWorkTask * task = new SimpleWorkTask(socket_client);
    gp_thread_pool->AddTask(task);
}

/**
 * 未采用的实现方式，备用
 */
void service()
{
    //创建server的listen socket
    int sock;
    if ( (sock = socket(PF_INET, SOCK_STREAM, 0)) == -1) 
    {
        perror("socket failed");
        return;
    }   

    //in case of 'address already in use' error message
    int yes = 1;
    if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int))) 
    {
        perror("setsockopt failed");
        return;
    }   

    //设置sock为non-blocking
    setSockNonBlock(sock);

    //创建要bind的socket address
    struct sockaddr_in bind_addr;
    memset(&bind_addr, 0, sizeof(bind_addr));
    bind_addr.sin_family = AF_INET;
    bind_addr.sin_addr.s_addr = htonl(INADDR_ANY);  //设置接受任意地址
    bind_addr.sin_port = htons(2046);

    //bind sock到创建的socket address上
    if ( bind(sock, (struct sockaddr *)&bind_addr, sizeof(bind_addr)) == -1 ) 
    {
        perror("bind failed");
        return;
    }

    //listen
    if (listen(sock, 5) == -1) 
    {
        perror("listen failed");
        return;
    }

    //初始化libevent
    struct event_base *p_base  = event_base_new();
    struct event * p_accept_ev = event_new(p_base,sock,EV_READ|EV_PERSIST,accept_callback, NULL);
    //将如上创建好的event添加到检测的事件中
    event_add(p_accept_ev, NULL);
    
    gp_event_base = p_base;
    //循环地检测已添加事件是否发生
    event_base_dispatch(p_base);

    shutdown(sock, SHUT_RDWR);
    event_free(p_accept_ev);
    gp_event_base = NULL;
    event_base_free(p_base);
}

/**
 * 返回服务器忙的信息
 */
int response_busynow(struct sockclient* socket_client)
{
    struct http_connection *conn = &socket_client->conn;
    conn->inbuf=NULL;
    conn->outbuf = evbuffer_new();
    const char * info ="service is too busy now!";
    set_http_response ( conn->outbuf,HTTP_OK,CODE_STR( HTTP_OK ) ,false, info, strlen(info) );
    evbuffer_write( conn->outbuf, conn->fd );
    evbuffer_free(conn->outbuf);
    conn->outbuf=NULL;
    disconnect_conn(&socket_client->conn);
    free(socket_client);
}

static void evli_accept_conn_cb(struct evconnlistener *listener, 
	evutil_socket_t fd, struct sockaddr *address, int socklen, void *ctx)
{
    DEBUG_LOG("%s:%s",__FILE__,__FUNCTION__);
    //保存client端信息到自定义结构体中
    struct sockclient * socket_client = (struct sockclient*)malloc(sizeof(struct sockclient));
    if (socket_client == NULL) 
	{
        WARN_LOG("malloc failed");
        shutdown(fd,SHUT_RDWR);
        return;
    }
    
    if (!inet_ntop(AF_INET, &(((struct  sockaddr_in *)address)->sin_addr), socket_client->ip, sizeof(socket_client->ip))) 
	{
        WARN_LOG("inet_ntop failed");
        shutdown(fd,SHUT_RDWR);
        return;
    }

    DEBUG_LOG("sock: %d, from: %s", fd, socket_client->ip);
    struct http_connection *conn = &socket_client->conn;
    conn->fd     = fd;
    conn->inbuf  = NULL;
    conn->outbuf = NULL;
    
    SimpleWorkTask * task = new SimpleWorkTask(socket_client);
    int res = gp_thread_pool->AddTask(task);
    if(res < 0) //添加失败，显示线程忙
    {
        WARN_LOG("server is busy now!");
        response_busynow(socket_client);
        delete task;
    }
}

static void evli_accept_error_cb(struct evconnlistener *listener, void *ctx)
{
    struct event_base *base = evconnlistener_get_base(listener);
    int err = EVUTIL_SOCKET_ERROR();
    ERROR_LOG("Got an error %d (%s) on the listener. Shutting down.",err, evutil_socket_error_to_string(err));
    event_base_loopexit(base, NULL);
}

void service_evlistener()
{
    struct sockaddr_in bind_addr;
    memset(&bind_addr, 0, sizeof(bind_addr));
    bind_addr.sin_family = AF_INET;
    bind_addr.sin_addr.s_addr = htonl(INADDR_ANY);  //设置接受任意地址
    bind_addr.sin_port = htons(2046);

    //初始化libevent
    struct event_base *p_base = event_base_new();
    gp_event_base = p_base;
    
    struct evconnlistener * listener = evconnlistener_new_bind(p_base, evli_accept_conn_cb, NULL,
            LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1, (struct sockaddr*)&bind_addr, sizeof(bind_addr));
    if (!listener) {
        perror("Couldn't create listener");
        return;
    }
    evconnlistener_set_error_cb(listener, evli_accept_error_cb);
    event_base_dispatch(p_base);
    
    evconnlistener_free(listener);
    gp_event_base = NULL;
    event_base_free(p_base);
}

/**
 * 单线程的服务,旧代码，备用
 */
void service_single()
{
    evthread_use_pthreads();
    //event_config *base_cfg=event_config_new();
    //event_config_set_flag(base_cfg,EVENT_BASE_FLAG_STARTUP_IOCP);//启动IOCP
    //event_config_set_num_cpus_hint(base_cfg,ncpucount);
    gp_event_base = event_base_new();
    struct evhttp * httpd = evhttp_new(gp_event_base);
    if (httpd == NULL) {
        ERROR_LOG("evhttp_new() ... failure");
        exit(1);
    }
    int res = evhttp_bind_socket(httpd,gp_parameter->host.c_str(),(uint16_t)gp_parameter->port);
    if(res != 0){
        ERROR_LOG("evhttp_bind_socket() ... failure %d",res);
        exit(1);
    }
    // 设置请求超时时间
    evhttp_set_timeout(httpd, 3);

    sphinx_client * const client = get_thread_sphinx_client();
    
    sphinx_set_server(client, gp_parameter->proxyed_host.c_str(), gp_parameter->proxyed_port);
    //gp_json_writer = new Json::StyledWriter;
    //gp_json_writer=new Json::FastWriter;

    cout << "start, waiting for search..." << endl;
    //evhttp_set_gencb(httpd, http_handler, NULL);
    //FIXME:
    //evhttp_set_gencb(httpd, dumb, NULL);
    res = event_base_dispatch(gp_event_base);
    if(res != 0){
        ERROR_LOG("event_base_dispatch() ... failure %d ",res);
    }
    evhttp_free(httpd);
    event_base_free(gp_event_base);
}

int main(int argc, char **argv)
{
    gp_parameter = new GlobalParameter;
	//配置文件解析
	if (!load_config_file()) 
    {
		fprintf(stderr,"加载配置文件失败\n");
		return 1;
	}
	//全部参数从配置文件读取,不再接受命令行参数
	check_config_parameter();
	//全局的日志
	gp_log = new CLog(gp_parameter->log_path_name.c_str());
    if(gp_parameter->log_level == "INFO")
    {
        gp_log->set_log_level(CLog::LOG_INFO);
    }
    else if(gp_parameter->log_level == "DEBUG")
    {
        gp_log->set_log_level(CLog::LOG_DEBUG);
    }
    else if(gp_parameter->log_level == "WARN")
    {
        gp_log->set_log_level(CLog::LOG_WARN);
    }
    else if(gp_parameter->log_level == "ERROR")
    {
        gp_log->set_log_level(CLog::LOG_ERROR);
    }
    else
    {
        gp_log->set_log_level(CLog::LOG_NO);
    }
	INFO_LOG("开始");

	if (gp_parameter->is_daemon) {
		if (fork() == 0 /* CHILD PROCESS */ ) {
			INFO_LOG("Daemon mode start.");
			cout << "Daemon mode start." << endl;
            setsid();
		} else {	//parent
			return 0;
		}
	}
    
    //for pid
    check_pid_file();
	SetSignalHandle();

	gp_statistic = new Statistic;
    gp_thread_pool = new CThreadPool(gp_parameter->thread_pool_min,gp_parameter->thread_pool_max);
    //gp_thread_pool->set_thread_exit_cb(release_sphinx_client);
    //gp_thread_pool->set_before_run_task_cb(thread_pool_before_run_task_cb);
    //调试使用
    gp_thread_pool->set_after_run_task_cb(release_sphinx_client);

    //service();
    service_evlistener();
    
    delet_pid_file();

	delete gp_statistic;
    delete gp_parameter;
    delete gp_thread_pool;

    cout << "main return!!!" << endl;
	INFO_LOG("结束");

    gp_log->close();
    delete gp_log;

	return 0;
}

/**
 * 单独测试Sphinx Client性能
 */
int main_test(int argc, char **argv)
{
    sphinx_client * const client = get_thread_sphinx_client();
    sphinx_set_server(client, "192.168.22.33",9312);
    sphinx_set_connect_timeout(client, 3.0);
    sphinx_open(client);
    //gp_json_writer = new Json::StyledWriter;

    QueryData qd;
    qd.index="mysql";
    qd.q="%E6%90%9C%E7%B4%A2";
    qd.rankingmode="SPH_RANK_PROXIMITY_BM25";
    qd.sortmode="SPH_SORT_RELEVANCE";
    qd.matchmode="SPH_MATCH_EXTENDED2";
    SphinxQueryData sqd(qd);
    for (int i=0;i<1000;i++)
    {
        SearchResult sr;
        search(sqd,&sr);
        string str= convert_to_json_string(sr); 
        if(i%100==0) cout<<i<<endl;
    }
    sphinx_close(client);
    return 0;
}

int main2(int argc, char ** argv)
{
    sphinx_client *clients[100];
    int len=0;
    int p=0;

    int LEN=100;
    for(int i=0;i<LEN;i++){
        printf("%d\n",i);
        sphinx_client * client = sphinx_create(SPH_TRUE);
        if(client == NULL) {
            perror("NULL");
            exit(2);
        }
        if( sphinx_set_server(client, "192.168.22.33",9312)
                &&
                sphinx_open(client) )
        {
            printf("ok");
        }
        else{
            printf("error");
            break;
        }

        if (sphinx_set_match_mode(client, SPH_MATCH_ALL) != SPH_TRUE
                || sphinx_set_ranking_mode(client,SPH_RANK_PROXIMITY_BM25) != SPH_TRUE
                || sphinx_set_sort_mode(client, SPH_SORT_RELEVANCE, "") != SPH_TRUE) 
        {
            printf("error:set mode");
            exit(33);
        }
        sphinx_result *res = sphinx_query(client, "mm","mysql", NULL);
        if(res==NULL){
            printf("error");
            exit(3);
        }
        clients[i]=client;
    }
    for(int i=0;i<LEN;i++){
        sphinx_destroy(clients[i]);
    }
}

